package com.cxl.kafka;

import cn.hutool.core.util.RandomUtil;
import com.alibaba.fastjson.JSON;
import com.cxl.entity.User;
import com.cxl.esrepository.EsRepository;
import com.cxl.service.IUserService;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.io.FileSystemResource;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.mail.javamail.JavaMailSenderImpl;
import org.springframework.mail.javamail.MimeMessageHelper;
import org.springframework.stereotype.Component;
import org.springframework.util.DigestUtils;

import javax.mail.MessagingException;
import javax.mail.internet.MimeMessage;
import java.io.File;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeUnit;

/**
 * @author: 20971
 * @time: 2022/3/11 16:52
 * @description:
 */
@Component
public class ConsumerListener {

    @Autowired
    private RedisTemplate redisTemplate;

    @Autowired
    private IUserService iUserService;

    @Autowired
    private EsRepository esRepository;

    @Autowired
    private JavaMailSenderImpl javaMailSender;

    @KafkaListener(topics = {"test"})
    public void KafkaConsumer(ConsumerRecord<String, String> consumerRecord, Acknowledgment acknowledgment) throws MessagingException {
        String key = consumerRecord.key();
        String value = consumerRecord.value();
        if (StringUtils.isNoneBlank(key) && StringUtils.isNoneBlank(value)) {
            //kafka防重复消费
            if (!redisTemplate.opsForValue().setIfAbsent(value, null, 1, TimeUnit.HOURS)) {
                if (key.equals("add")) {
                    //MySql添加 + es同步
                    User user = JSON.parseObject(value, User.class);
                    String pwd = RandomUtil.randomNumbers(4);
                    pwd = DigestUtils.md5DigestAsHex(pwd.getBytes(StandardCharsets.UTF_8));
                    user.setPwd(pwd);
                    boolean save = iUserService.save(user);
                    if (save) {
                        //发送添加用户成功邮箱
                        MimeMessage mimeMessage = javaMailSender.createMimeMessage();
                        MimeMessageHelper helper = new MimeMessageHelper(mimeMessage, true);
                        helper.setFrom("2097113175@qq.com");
                        helper.setTo(user.getEmail());
                        helper.setSubject("添加用户");
                        helper.setReplyTo("回复不限");
                        helper.setText("<html><body><font color='red'>" + user.toString() + "</font></body><html>", true);
                        FileSystemResource file = new FileSystemResource(new File("D:\\pic\\" + user.getImg()));
                        javaMailSender.send(mimeMessage);
                    }
                    //es同步
                    esRepository.save(user);
                }
            }
            redisTemplate.delete(value);
        }
        //手动提交
        acknowledgment.acknowledge();
    }
}
